package com.hu.flink12.api.state;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author: hujianjun
 * @Create Date: 2021/2/7 9:44
 * @Describe: 求最大值，可以直接使用keyby().maxby()直接实现，也可以通过底层state实现
 * max和maxby之间虽然都能返回整个元素，但是max除了获取每次比较时最大的元素，但其他位置的元素是第一次获取到的
 *
 */
public class MaxByState {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //TODO 2.读取数据源
        DataStream<Tuple3<String, Integer, Integer>> inputStream = env.fromElements(
                Tuple3.of("bj", 2, 2),
                Tuple3.of("sh", 1, 1),
                Tuple3.of("bj", 5, 6),
                Tuple3.of("bj", 3, 5),
                Tuple3.of("bj", 1, 9),
                Tuple3.of("bj", 2, 8),
                Tuple3.of("bj", 3, 10),
                Tuple3.of("sh", 2, 9)
        );

        //TODO 3.转换操作
        DataStream<Tuple3<String, Integer, Integer>> maxResult = inputStream
                .keyBy(t -> t.f0)
                /**
                 * 2> (sh,1,1)
                 * 2> (sh,1,9)
                 * 7> (bj,2,2)
                 * 7> (bj,2,6)
                 * 7> (bj,2,6)
                 * 7> (bj,2,9)
                 * 7> (bj,2,9)
                 * 7> (bj,2,10)
                 */
//                .max(2);
                /**
                 * 2> (sh,1,1)
                 * 7> (bj,2,2)
                 * 2> (sh,2,9)
                 * 7> (bj,5,6)
                 * 7> (bj,5,6)
                 * 7> (bj,1,9)
                 * 7> (bj,1,9)
                 * 7> (bj,3,10)
                 */
                .maxBy(2);

        //TODO 4.sink操作
        maxResult.print();

        //TODO 5.执行程序
        env.execute();
    }
}
